/******************************************************************************
 * Copyright (C) 671643387@qq.com
 *
 * This program is free software; you can redistribute it and/or modify it
 * under the terms of the GNU General Public License as published by the
 * Free Software Foundation; either version 2 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
 * more details.
 *
 * You should have received a copy of the GNU General Public License along
 * with this program; if not, write to the Free Software Foundation, Inc.,
 * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
 *
 *****************************************************************************/

#include "NodeCpp/Net/TcpServer.h"
#include <cassert>
#include <cstring>
#include <memory>
#include "NodeCpp/Error.h"
#include "NodeCpp/Net/SocketApi.h"
#include "NodeCpp/Net/TcpSocketDelegateFactory.h"
#include "NodeCpp/Net/TcpSocket.h"

namespace NodeCpp {
namespace Net {

    TcpRunnable::TcpRunnable(void)
        : Base_(NULL)
        , DummyEvent_(NULL)
        , Load_(0)
    {

    }

    TcpRunnable::~TcpRunnable(void)
    {
        _Uninitialize();
    }

    bool TcpRunnable::Init(void)
    {
        do
        {
            Base_ = event_base_new();
            if (!Base_) {
                break;
            }

            DummyEvent_ = event_new(Base_, -1, EV_TIMEOUT | EV_PERSIST,
                &TcpRunnable::_OnTimeout, this);
            if (!DummyEvent_) {
                break;
            }
            struct timeval tv = { 1, 0 };
            if (0 != event_add(DummyEvent_, &tv)) {
                break;
            }

            return true;
        } while (false);
        return false;
    }

    void TcpRunnable::Stop(void)
    {
        event_base_loopbreak(Base_);
    }

    void TcpRunnable::Run(void)
    {
        event_base_dispatch(Base_);
    }

    void TcpRunnable::_Uninitialize(void)
    {

        if (DummyEvent_ != NULL) {
            event_free(DummyEvent_);
            DummyEvent_ = NULL;
        }
        if (Base_ != NULL) {
            event_base_free(Base_);
            Base_ = NULL;
        }
    }

	void TcpRunnable::_OnTimeout(evutil_socket_t, short, void *)
    {

    }


    const static float MPA = 0.9f;

    TcpServer::TcpServer(TcpSocketDelegateFactory *_ConnFactory /* = NULL */)
        : Acceptor_(NULL)
        , ConnIDGen_(0)
        , Connections_()
        , ConnectionsLock_()
        , Connf_(_ConnFactory)
        , MainWorker_()
        , AddtionalWorkerNum_(0)
        , AddtionalWorkers_()
        , ErrorCode_(0)
        , ErrorString_()
        , RunnerPool_(NULL)
        , RunnerPriority_(Thread::PRIO_NORMAL)
        , Running_(false)
    {

    }

    TcpServer::~TcpServer(void)
    {
        _Uninitialize();
    }

    void TcpServer::SetConnectionFactory(TcpSocketDelegateFactory *_ConnFactory)
    {
        NODECPP_ASSERT(!Connf_);
        Connf_.reset(_ConnFactory);
    }

    bool TcpServer::SetThreadPool(ThreadPool* _ThreadPool, Thread::Priority _Priority /* = Thread::PRIO_NORMAL */)
    {
        NODECPP_ASSERT(RunnerPool_ == NULL);
        RunnerPool_ = _ThreadPool;
        RunnerPriority_ = _Priority;

        if (Acceptor_) {
            if (!_StartRunner()) {
                return false;
            }
            Running_ = true;
        }
        return true;
    }

    bool TcpServer::Start(const char *_Host, int _Port, int _Backlog, int _AddWorkerNum /* = 0 */)
    {
        NODECPP_ASSERT(Connf_);
        AddtionalWorkerNum_ = _AddWorkerNum;

        do
        {
            if (!MainWorker_.Init()) {
                break;
            }

            Acceptor_.reset(new TcpAcceptor(MainWorker_.AcquireBase(), 
                &TcpServer::_OnAccept, this));
            if (!(Acceptor_ && Acceptor_->Open(_Host, _Port, _Backlog))) {
                break;
            }

            if (_AddWorkerNum > 0 && !_CreateAddtionalWorkers(_AddWorkerNum)) {
                break;
            }

            if (RunnerPool_ != NULL) {
                if (!_StartRunner()) {
                    break;
                }
                Running_ = true;
            }
            return true;
        } while (false);

        _GenErrorInfo();
        return false;
    }

    void TcpServer::Stop(void)
    {
        Running_ = false;

        /* Stop listener */
        Acceptor_->Close();

        /* Stop addtional workers */
        _StopAddtionalWorkers();

        /* Stop main worker */
        MainWorker_.Stop();

        /* Close all connections */
        {
            NODECPP_LOCK_MUTEX(_Guard, ConnectionsLock_);
            SocketMap::iterator _Iter = Connections_.begin();
            for (; _Iter != Connections_.end(); ++_Iter)
            {
                ScopedPtr<TcpSocket> _Sock(_Iter->second);
                _Sock->Delegate_->OnClose();
            }
            Connections_.clear();
        }
    }

    void TcpServer::_Uninitialize(void)
    {
        _CloseAddtionalWorkers();
    }

    void TcpServer::_GenErrorInfo(void)
    {
        ErrorCode_ = NodeCpp::GetLastError();
        NodeCpp::GetErrorString(ErrorString_, ErrorCode_);
    }

    void TcpServer::_CloseSocket(TcpSocket *_Sock)
    {
        NODECPP_ASSERT(_Sock != NULL);
        {
            NODECPP_LOCK_MUTEX(_Guard, ConnectionsLock_);
            SocketMap::iterator _Iter = Connections_.find(_Sock->id());
            NODECPP_ASSERT(_Iter != Connections_.end());
            if (_Iter != Connections_.end())
            {
                ScopedPtr<TcpSocket> _Sock(_Iter->second);
                _Sock->Delegate_->OnClose();
                _Sock->Runnable_->ReleaseBase();
                Connections_.erase(_Iter);
            }
        }
    }

    bool TcpServer::_CreateAddtionalWorkers(int _AddWorkerNum)
    {
        do
        {
            AddtionalWorkers_.reset(new TcpRunnable *[MAX_WORKERS]);
            if (!AddtionalWorkers_) {
                return false;
            }
            memset(AddtionalWorkers_.get(), 0, sizeof(TcpRunnable*)* MAX_WORKERS);

            int _NumWorkers = 0;
            for (; _NumWorkers < _AddWorkerNum; ++_NumWorkers)
            {
                AutoPtr<TcpRunnable> _Worker(new TcpRunnable());
                if (!_Worker.get()) {
                    return false;
                }

                if (_Worker->Init()) {
                    AddtionalWorkers_[_NumWorkers] = _Worker.release();
                    continue;
                }
            }
        } while (false);

        return true;
    }

    void TcpServer::_CloseAddtionalWorkers(void)
    {
        if (AddtionalWorkerNum_ > 0 && AddtionalWorkers_)
        {
            for (std::size_t _X = 0; _X < AddtionalWorkerNum_; ++_X)
            {
                ScopedPtr<TcpRunnable> _Runnable(AddtionalWorkers_[_X]);
                if (_Runnable) {
                    _Runnable->Stop();
                    AddtionalWorkers_[_X] = NULL;
                }
            }
            AddtionalWorkers_.reset();
        }
    }

    void TcpServer::_StopAddtionalWorkers(void)
    {
        if (AddtionalWorkerNum_ > 0 && AddtionalWorkers_)
        {
            for (std::size_t _X = 0; _X < AddtionalWorkerNum_; ++_X)
            {
                if (AddtionalWorkers_[_X] != NULL) {
                    AddtionalWorkers_[_X]->Stop();
                }
            }
        }
    }

	void TcpServer::_OnAccept(evutil_socket_t _Sock, struct sockaddr *_SockAddr, int _AddrLen, void *_Userptr)
    {
        TcpServer *_Server = static_cast<TcpServer*>(_Userptr);
        NODECPP_ASSERT(_Server);
        _Server->_DoAccept(_Sock, _SockAddr, _AddrLen, _Userptr);
    }

	void TcpServer::_DoAccept(evutil_socket_t _Sock, struct sockaddr * _SockAddr, int _AddrLen, void *)
    {
        do
        {
            NODECPP_LOCK_MUTEX(_Guard, ConnectionsLock_);

            /* Find a connection id */
            int _ConnId = ++ConnIDGen_;
            if (Connections_.find(_ConnId) != Connections_.end()) {
                break;
            }

            /* Accept the connection */
            AutoPtr<TcpSocket> _NewSock(new TcpSocket());
            AutoPtr<TcpSocketDelegate> _SockDelegate(Connf_->Create());
            if (_NewSock.get() && _SockDelegate.get())
            {
                TcpRunnable *_Worker = _FindWorker();
                if (_NewSock->Open(_Worker->AcquireBase(), _Sock, _SockAddr, _AddrLen))
                {
                    _SockDelegate->_SetPeer(_NewSock.get());
                    _NewSock->_SetId(_ConnId);
                    _NewSock->_SetOwner(this);
                    _NewSock->_SetThread(_Worker);
                    _NewSock->_SetDelegate(_SockDelegate.release());
                    _NewSock->Delegate_->OnOpen();
                    Connections_[_ConnId] = _NewSock.release();
                    return;
                }
            }

            /* Release the connection id */
            --ConnIDGen_;
        } while (false);

        CloseSocket(_Sock);
    }

    TcpRunnable *TcpServer::_FindWorker(void)
    {
        if (AddtionalWorkerNum_ > 0)
        {
            std::size_t _WorkerIdx = 0;
            std::size_t _MinLoad = UINT_MAX;

            /* Find most free attional worker */
            for (std::size_t _X = 0; _X < AddtionalWorkerNum_; ++_X)
            {
                std::size_t _WorkerLoad = AddtionalWorkers_[_X]->GetLoad();
                if (_WorkerLoad < _MinLoad)
                {
                    _MinLoad = _WorkerLoad;
                    _WorkerIdx = _X;
                }
            }

            std::size_t _MainLoad = MainWorker_.GetLoad();
            if (_MainLoad >(std::size_t)(_MinLoad * MPA)) {
                return AddtionalWorkers_[_WorkerIdx];
            }
        }

        return &MainWorker_;
    }

    bool TcpServer::_StartRunner(void)
    {
        NODECPP_ASSERT(RunnerPool_ != NULL);
        if (!RunnerPool_->Start(&MainWorker_, RunnerPriority_)) {
            return false;
        }

        for (std::size_t _X = 0; _X < AddtionalWorkerNum_; ++_X)
        {
            if (!RunnerPool_->Start(AddtionalWorkers_[_X], RunnerPriority_)) {
                return false;
            }
        }

        return true;
    }
}}
